package cn.doitedu.feature_etl;

import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class Job01_Ad_click_predict_features {

    public static void main(String[] args) {


        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.enableCheckpointing(5000, CheckpointingMode.EXACTLY_ONCE);
        env.getCheckpointConfig().setCheckpointStorage("file:///d:/ckpt");

        env.setParallelism(1);


        StreamTableEnvironment tenv = StreamTableEnvironment.create(env);
        tenv.getConfig().set("table.exec.source.idle-timeout","1 ms");



        // 1. 创建kafka数据源表（行为日志明细宽表）
        tenv.executeSql(
                "create table dwd_events_kafka(     " +
                        "     user_id bigint             " +
                        "    ,event_id           string             " +
                        "    ,action_time        bigint             " +
                        "    ,properties         map<string,string> " +
                        "    ,ad_tracking_id as properties['ad_tracking_id'] " +
                        "    ,ad_id as properties['ad_id']" +
                        "    ,rt as to_timestamp_ltz(action_time,3)   " +
                        "    ,watermark for rt as rt - interval '0' second   " +
                        ") WITH (                      " +
                        "    'connector' = 'kafka',    " +
                        "    'topic' = 'dwd-user-action-log',  " +
                        "    'properties.bootstrap.servers' = 'doitedu:9092',  " +
                        "    'properties.group.id' = 'doit44_g1',              " +
                        "    'scan.startup.mode' = 'latest-offset',            " +
                        "    'value.format' = 'json',                          " +
                        "    'value.fields-include' = 'EXCEPT_KEY'             " +
                        ")                                                     ");


        tenv.executeSql(
                "create temporary view show_click as " +
                "with tmp as ( \n" +
                "    select \n" +
                "    *\n" +
                "    from dwd_events_kafka\n" +
                "    where event_id in ('ad_show','ad_click','ad_transform')\n" +
                ")\n" +
                "select\n" +
                "    uid,\n" +
                "    adid,\n" +
                "    tkid,\n" +
                "    show_time,      \n" +
                "    click_time,     \n" +
                "    transform_time, \n" +
                "    proctime() as pt  \n" +
                "from tmp \n" +
                "MATCH_RECOGNIZE(\n" +
                "    PARTITION BY ad_tracking_id\n" +
                "    ORDER BY rt \n" +
                "    MEASURES\n" +
                "       A.user_id as uid,\n" +
                "       A.action_time as show_time,\n" +
                "       A.ad_id as adid,\n" +
                "       A.ad_tracking_id as tkid,\n" +
                "       first(B.action_time) as click_time,\n" +
                "       C.action_time as transform_time\n" +
                "    \n" +
                "    ONE ROW PER MATCH \n" +
                "    AFTER MATCH SKIP PAST LAST ROW \n" +
                "    PATTERN(A B+ C)\n" +
                "    DEFINE\n" +
                "       A AS A.event_id = 'ad_show',\n" +
                "       B AS B.event_id = 'ad_click',\n" +
                "       C AS C.event_id = 'ad_transform'\n" +
                ") ").print();


        // 建表，映射hbase中的广告引擎后端请求特征日志数据表
        tenv.executeSql(
                " create table lookup_hbase (                      "+
                        "   ad_tracking_id string                 "+
                        "   ,f ROW<features string>               "+
                        " ) WITH (                                "+
                        "  'connector' = 'hbase-2.2',             "+
                        "  'table-name' = 'ad_request_log',         "+
                        "  'zookeeper.quorum' = 'doitedu:2181'    "+
                        " )                                        "
        );



        // 用曝光点击转化流，去lookup关联这个  请求特征日志表

        tenv.executeSql(
                "select\n" +
                " a.uid \n" +
                ",a.adid \n" +
                ",a.tkid \n" +
                ",a.show_time \n" +
                ",a.click_time \n" +
                ",a.transform_time \n" +
                ",b.features\n" +
                "\n" +
                "from show_click a left join lookup_hbase  \n" +
                "for system_time as of a.pt   as b on a.tkid = b.ad_tracking_id").print();

    }


}
